package com.apps.sdses.flink141.cdc;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/22 8:53
 */
public class MySQLCDC2Kafka {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setStateBackend(new HashMapStateBackend());
        env.enableCheckpointing(10, CheckpointingMode.EXACTLY_ONCE);


        //source
        tableEnv.executeSql("CREATE TABLE products (\n" +
                "    id INT NOT NULL,\n" +
                "    name STRING,\n" +
                "    description STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'server-time-zone' = '+08:00',\n" +
                "    'database-name' = 'mydb',\n" +
//                "    'scan.incremental.snapshot.enabled' = 'false',\n" +
//                "    'server-id' = '5401-5405',\n" +
                "    'table-name' = 'products'\n" +
                "  )");

        //sink
//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE products (EXCLUDING ALL)");

        tableEnv.executeSql("CREATE temporary view source_v as select max(id) from products");


        tableEnv.executeSql("CREATE TABLE sinkTable(id INT) WITH ('connector' = 'print')");

        tableEnv.executeSql("insert into sinkTable select * from source_v ");


//        tableEnv.executeSql("CREATE TABLE sinkTable (" +
//                "    id BIGINT,\n" +
//                "    name STRING,\n" +
//                "    description STRING,\n" +
//                "    PRIMARY KEY (id) NOT ENFORCED \n" +
//                "   )\n" +
//                "   WITH (\n" +
//                "  'connector' = 'kafka',\n" +
//                "  'topic' = 'test_auto_create_topic_01',\n" +
//                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
//                "  'properties.allow.auto.create.topics' = 'true',\n" +
////                "  'format' = 'canal-json'\n" +
////                "  'format' = 'maxwell-json'\n" +
//                "  'format' = 'debezium-json'\n" +
////                "  'key.format' = 'json',\n" +               //json格式序列化:{"id":1,"sid":"11baac5a","ts":"2023-06-12 15:09:22.849","flag":true,"vc":31}
////                "  'value.format' = 'json'\n" +               //json格式序列化:{"id":1,"sid":"11baac5a","ts":"2023-06-12 15:09:22.849","flag":true,"vc":31}
//                ")");

        //insert
//        tableEnv.executeSql("insert into sinkTable select * from products ");
    }
}
